package com.shujia.core

import org.apache.spark.rdd.RDD
import org.apache.spark.util.LongAccumulator
import org.apache.spark.{SparkConf, SparkContext}

/**
 * 累加器
 *
 * spark写代码的注意事项
 * 1、算子内部可以使用外部定义变量，但是算子外部无法直接使用算子内部的变量值，使用累加器解决
 * 2、spark的RDD不能嵌套使用
 * 3、算子内部无法使用SparkContext对象
 *
 */
object AccumulatorDemo {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
    conf.setMaster("local")
    conf.setAppName("累加器")
    val sc = new SparkContext(conf)
    //==================================================


    //读取一个文件，统计其中的数据条数
//    var count = 0
//    val lineRDD: RDD[String] = sc.textFile("spark/data/students.txt")
//    lineRDD.foreach((e: String) => {
//      count += 1
//      println(s"count:$count")
//      println("--------------------")
//    })
//    println(s"该文件的数据条数为: ${count}")


    //使用累加器代替上面的统计变量。一种是整数累加器，一种是小数累加器
//    val acc1: LongAccumulator = sc.longAccumulator
//    val lineRDD: RDD[String] = sc.textFile("spark/data/students.txt")
//    lineRDD.foreach((e: String) => {
//      acc1.add(1)
//      println(s"acc1:${acc1.value}")
//      println("--------------------")
//    })
//    println(s"该文件的数据条数为: ${acc1.value}")



    //spark的RDD不能嵌套使用
//    val lineRDD: RDD[String] = sc.textFile("spark/data/students.txt")
//    val lineRDD2: RDD[String] = sc.textFile("spark/data/score.txt")
//    val value2: RDD[RDD[(String, String)]] = lineRDD.map((s: String) => {
//      lineRDD2.map((e: String) => {
//        val id = s.split(",")(0)
//        val score = e.split(",")(2)
//        (id, score)
//      })
//    })
//
//    value2.foreach(println)

    //算子内部无法使用SparkContext对象
//    val lineRDD: RDD[String] = sc.textFile("spark/data/students.txt")
//    val v2: RDD[String] = lineRDD.map((e: String) => {
//      val lineRDD2: RDD[String] = sc.textFile("spark/data/score.txt")
//      e
//    })
//
//    v2.foreach(println)


  }
}
